package com.tang.sink;

import com.tang.deserialization.CustomDeserialization;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.JsonDebeziumSchemaSerializer;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;
import java.util.StringJoiner;
import java.util.UUID;

/**
 * Mysql CDC到 Doris
 *
 * @author tang
 */
public class MySqlSInk2Doris {

    public static void main(String[] args) throws Exception {

        /*Map<String, Object> customConverterConfigs = new HashMap<>();
        customConverterConfigs.put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, "numeric");
        JsonDebeziumDeserializationSchema schema =
                new JsonDebeziumDeserializationSchema(false, customConverterConfigs);*/

        String databaseName = "lottery";
        int databaseCount = 2, databaseStart = 1;
        String tableName = "user_strategy_export";
        int tableCount = 4, tableStart;

        StringJoiner databaseJoin = new StringJoiner(",");
        StringJoiner tableJoin = new StringJoiner(",");
        for (int i = 0; i < databaseCount; i++) {
            String dataBaseName = databaseName + "_0" + databaseStart++;
            databaseJoin.add(dataBaseName);
            tableStart = 0;
            for (int i1 = 0; i1 < tableCount; i1++) {
                tableJoin.add(dataBaseName + "." + tableName + "_00" + tableStart++);
            }
        }

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("192.168.70.141")
                .port(3306)
                .databaseList(databaseJoin.toString()) // set captured database
                .tableList(tableJoin.toString()) // set captured table
                .username("root")
                .password("123@abAB")
                .serverTimeZone("UTC") // 注意看报错，可能会提示时区不一致问题
                .deserializer(new CustomDeserialization()) // converts SourceRecord to JSON String
                .build();

        //StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // enable checkpoint
        env.enableCheckpointing(10000);
//
        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("read_json_by_line", "true");

        DorisOptions dorisOptions = DorisOptions.builder()
                .setFenodes("192.168.70.141:8030")
                .setTableIdentifier("lottery.user_strategy_export")
                .setUsername("root")
                .setPassword("").build();
//
        DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
        executionBuilder.setLabelPrefix("label-doris" + UUID.randomUUID())
                .setStreamLoadProp(props).setDeletable(true);

        DorisSink.Builder<String> builder = DorisSink.builder();
        builder.setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisExecutionOptions(executionBuilder.build())
                .setDorisOptions(dorisOptions)
                .setSerializer(JsonDebeziumSchemaSerializer.builder().setDorisOptions(dorisOptions).build());

        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")//.print();
                .sinkTo(builder.build());

        env.execute("Print MySQL Snapshot + Binlog");
    }
}